package com.gxlevi.hdfs.util;


import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.io.*;

/**
 * HDFS基本操作工具类
 */
public class HDFSUtil {
    //伪装用户
    private String user = "root";
    //hdfs地址
    private String url = "hdfs://node01:8020";
    private FileSystem fileSystem = null;
    private Configuration conf = new Configuration();

    public void init() throws URISyntaxException, IOException, InterruptedException {
        if ("".equals(user) || null == user) {
            fileSystem = FileSystem.get(new URI(url), conf);
        } else {
            fileSystem = FileSystem.get(new URI(url), conf, user);
        }
    }

    public HDFSUtil() throws InterruptedException, IOException, URISyntaxException {
        init();
    }

    /**
     * 检查文件或者文件夹是否存在
     *
     * @param fileName 文件路径
     * @return
     */
    public boolean checkFileExist(String fileName) {
        Path path = new Path(fileName);
        try {
            return fileSystem.exists(path);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * 创建文件夹
     *
     * @param dir 文件目录
     * @return
     */
    public boolean mkdirs(String dir) {
        if (checkFileExist(dir)) {
            return true;
        }
        try {
            return fileSystem.mkdirs(new Path(dir));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * hdfs文件下载到本地
     *
     * @param hdfsPath  hdfs文件路径
     * @param localPath 本地路径
     * @return
     */
    public boolean get(String hdfsPath, String localPath) {
        try {
            fileSystem.copyToLocalFile(new Path(hdfsPath), new Path(localPath));
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 本地文件上传
     *
     * @param hdfsFlie  hdfs文件路径
     * @param localPath 本地路径
     * @return
     */
    public boolean put(String hdfsFlie, String localPath) {
        try {
            fileSystem.copyFromLocalFile(new Path(localPath), new Path(hdfsFlie));
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除文件
     *
     * @param file 文件目录
     * @param r    是否递归删除
     * @return
     */
    public boolean rm(String file, boolean r) {
        try {
            fileSystem.delete(new Path(file), r);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 查看文件内容
     *
     * @param path 文件路径
     * @return
     * @throws IOException
     */
    public String cat(String path) throws IOException {
        if (!checkFileExist(path)) {
            System.out.println("文件不存在");
            return null;
        }
        FSDataInputStream inputStream = fileSystem.open(new Path(path));
        StringWriter writer = new StringWriter();
        IOUtils.copy(inputStream, writer);
        return writer.toString();
    }

    /**
     * 合并小文件
     *
     * @param mergeFile hdfs合并后的文件名
     * @param localDir  本地小文件集合所在目录
     * @throws IOException
     */
    @Test
    public void mergeFile(String mergeFile, String localDir) throws IOException {
        FSDataOutputStream outputStream = fileSystem.create(new Path(mergeFile));
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        FileStatus[] fileStatuses = localFileSystem.listStatus(new Path(localDir));

        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream = localFileSystem.open(fileStatus.getPath());
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        IOUtils.closeQuietly(outputStream);
        localFileSystem.close();
    }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        fileSystem.close();
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public FileSystem getFileSystem() {
        return fileSystem;
    }

    public void setFileSystem(FileSystem fileSystem) {
        this.fileSystem = fileSystem;
    }
}
